package com.crazymaker.cloud.nacos.demo.gateway.service;

import com.crazymaker.springcloud.common.constants.SessionConstants;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import com.crazymaker.springcloud.distribute.idGenerator.impl.IdWorker;
import com.crazymaker.springcloud.distribute.idGenerator.impl.ZkSequenceIdGenerator;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * rocketMQ中继队列的发送service
 * 实现了 ApplicationContextAware接口, setApplicationContext 方法
 * 以便spring在初始化上下文是将MQ生产者客户端创建实例和初始化
 * 注意 setApplicationContext 只能被调用一次, 否则会造成消息生产者创建的混乱
 * startMutiSender 方法是加锁的
 */
@Component
@Slf4j
public class RocketmqMessageRelayService implements ApplicationContextAware {
    /**
     * 消息大小限制, 4M
     */
    public static final int SIZE_LIMIT = 1024 * 1024 * 4;


    @Value("${rocketmq.nameServer}")
    private String rocketmqAddress;

    /**
     * 生产者中继队列分组
     */
    @Value("${rocketmq.producer.relayGroup}")
    private String producerRelayGroup;

    /**
     * 中继队列的topic
     */
    @Value("${rocketmq.producer.relayTopic}")
    private String relayTopic;

    /**
     * 生产者并发量, 提高此参数可增加与broker连接的channel数量, 提高吞吐量
     */
    @Value("${rocketmq.producer.concurrent}")
    private int concurrent = 1;

    @Value("${rocketmq.producer.retryTimes}")
    private int retryTimes = 1;

    /**
     * 阻塞队列长度限制
     */
    @Value("${rocketmq.producer.maxWaited}")
    private int maxWaited = 100000;


    private long workerId;

    List<InnerSender> senders = null;

    /**
     * 阻塞队列, 用于汇总待发送的消息
     */
    LinkedBlockingQueue<Message> messagesBlockList = new LinkedBlockingQueue<>();

    /**
     * id生成器, 默认用的zk生成, 也可以在实体类的id字段上指定
     */
    ZkSequenceIdGenerator msgIdGenerator = null;

    Executor executor = null;

    /**
     * 被使用的生产者计数
     */
    private AtomicInteger producerInUseCount = new AtomicInteger(0);
    private volatile boolean inited = false;

    /**
     * 进行rocketMQ的延迟消息转发
     *
     * @param messageBody
     * @param userId
     * @return
     */
    public Long putMessage(byte[] messageBody, String userId) {

        // 如果阻塞队列长度超过最大值, 说明消息发送有问题没执行完, 返回空让调用方等待或者自行处理
        if (messagesBlockList.size() > maxWaited) {
            log.error("缓存超出阈值，请稍后再发");
            return null;
        }

        // 生成消息的id
        Long id = msgIdGenerator.nextId();

        // 构建消息, 发到delay接力队列
        Message message = new Message(relayTopic /* Topic */,
                "toReplace" /* Tag */,
                String.valueOf(id),
                messageBody);

        // 消息中放置用户的userId用于鉴权
        message.putUserProperty(SessionConstants.USER_IDENTIFIER, userId);

        // 阻塞队列中添加消息
        messagesBlockList.add(message);

        // 获取当前被使用的生产者计数
        int index = producerInUseCount.get();
        // 如果计数超过了配置的并发数, 则不进行发送线程的获取和消息发送, 消息放在阻塞队列中等待后续发送
        if (index >= concurrent) {
            // 返回消息id
            return id;
        }

        // 未超过并发数, 进行消息发送线程的获取和发送
        // 计数加1
        producerInUseCount.incrementAndGet();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                // 执行批量消息发送
                doSend(index);
                // 计数减1
                producerInUseCount.decrementAndGet();
            }
        });
        // 返回消息id
        return id;
    }

    /**
     * 执行批量消息发送
     *
     * @param index
     */
    private void doSend(int index) {

        if (index < 0) index = 0;

        InnerSender sender = senders.get(index);
        Message message = null;
        boolean waitToLong = false;
        // 保存批量发送的消息的list
        List<Message> batchList = new LinkedList<>();
        int totalSize = 0;

        //死循环
        while (!waitToLong) {

            try {
                // 阻塞队列, 获取到消息就马上返回, 获取不到消息, 最多等待200毫秒, 然后返回空
                message = messagesBlockList.poll(200, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
                waitToLong = true;
            }

            // 如果没拿到消息, 但是阻塞队列不为空, 也进行发送
            if (waitToLong || null == message) {
                if (batchList.size() > 0) {
                    // 同步发送批量消息
                    sender.syncSend(batchList);
                    log.info("批量发送，一次发送{} 条消息", batchList.size());
                    batchList.clear();
                    batchList = null;
                }
                break;
            }


            // topic +body的大小
            int currentSize = sender.relayTopicLength + message.getBody().length;

            // peperties, 消息里面参数的大小
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                currentSize += entry.getKey().length() + entry.getValue().length();
            }

            // 增加rocketmq日志的开销20字节
            currentSize = currentSize + 20;

            // 统计单个消息的全部大小, 如果超出4M上限, 不进行消息发送
            if (currentSize > SIZE_LIMIT) {
                // 单个消息超过了最大的限制
                // 忽略,否则会阻塞分裂的进程
                log.error("消息太大");
//                sender.asyncSend(message);
                continue;
            }

            // 保存批量发送的消息的list不为空且超出4M, 进行消息发送
            if (batchList.size() > 0 && currentSize + totalSize > SIZE_LIMIT) {
                // 同步发送批量消息
                sender.syncSend(batchList);

                // 保存批量发送的消息的list清空
                batchList.clear();
                totalSize = 0;
            }

            // 还未超出4M, 仅进行list的添加
            batchList.add(message);
            totalSize += currentSize;


        }
    }

    /**
     * 封装了DefaultMQProducer 的内部类
     */
    @Data
    static class InnerSender {

        private String relayTopic;
        private String rocketmqAddress;
        private String producerGroup;
        private String topic;
        private int relayTopicLength;
        private String instanceName;


        DefaultMQProducer mqProducer = null;

        private int retryTimes = 3;

        /**
         * 异步发送消息
         *
         * @param msg
         */
        public void asyncSend(Message msg) {
            try {
                // SendCallback接收异步返回结果的回调
                mqProducer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("send ok,instance： {} ,message id ：{}", instanceName, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("send error,instance： {} ,error is ：{}", instanceName, e.getMessage());
                        e.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 同步发送
         *
         * @param msgs
         */
        public void syncSend(Collection<Message> msgs) {
            try {
                // 不设置SendCallback接收异步返回结果的回调, 该方法会阻塞当前线程，直到收到发送消息的响应，或抛出异常
                mqProducer.send(msgs);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 启动MQ生产者客户端
         */
        private void startMqProducer() {
            mqProducer = new DefaultMQProducer(producerGroup);
            //指定NameServer地址
            mqProducer.setNamesrvAddr(rocketmqAddress); //修改为自己的
            mqProducer.setRetryTimesWhenSendFailed(retryTimes);
            mqProducer.setInstanceName(instanceName);
            mqProducer.setProducerGroup(producerGroup);
            /**
             * Producer对象在使用之前必须要调用start初始化，初始化一次即可
             * 注意：切记不可以在每次发送消息时，都调用start方法
             */
            try {
                mqProducer.start();
                log.info("product {} start ...", instanceName);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }


    }

    /**
     * 创建多个消息生产者客户端
     */
    private synchronized void startMutiSender() {

        if (inited) return;
        inited = true;

        // 生产者消息发送的线程池, 保证主线程和消息发送线程是隔离的
        // 核心线程数和最大线程数是一致的, 不会有线程被关闭
        // 阻塞队列, 10000
        executor = new ThreadPoolExecutor(
                concurrent,
                concurrent,
                ThreadUtil.KEEP_ALIVE_SECONDS,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue(ThreadUtil.QUEUE_SIZE),
                new ThreadUtil.CustomThreadFactory("MqProducer"));
        ;


        senders = new ArrayList<>(concurrent);
        // 根据并发数进行MQ生产者客户端组的创建
        for (int i = 0; i < concurrent; i++) {
            // 构造 DefaultMQProducer 的封装类
            InnerSender sender = new InnerSender();
            senders.add(sender);
            // instanceName的构成, 决定了生产者客户端和broker之间关联channel的数量
            sender.setInstanceName(producerRelayGroup + "-" + workerId + "-" + i);
            sender.setProducerGroup(producerRelayGroup);
            sender.setRocketmqAddress(rocketmqAddress);
            sender.setRetryTimes(retryTimes);
            sender.setRelayTopic(relayTopic);
            sender.setRelayTopicLength(sender.getRelayTopic().getBytes().length);

            // 启动MQ生产者客户端
            sender.startMqProducer();

        }

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        IdWorker idWorker = IdWorker.getInstance();
        workerId = idWorker.getWorkId();

        msgIdGenerator = new ZkSequenceIdGenerator("MsgPO");
        startMutiSender();
    }
}
